package cn._51doit.live.test;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 完成WindowTopN的官方例子
 */
public class WindowTopNDemo {


    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //创建Kafka的Source表
        //{"bidtime": "2020-04-15 08:05:00.000", "item" : "A", "price": 4, "supplier_id": "supplier1"}
        //{"bidtime": "2020-04-15 08:06:00.000", "item" : "C", "price": 4, "supplier_id": "supplier2"}
        //{"bidtime": "2020-04-15 08:07:00.000", "item" : "G", "price": 2, "supplier_id": "supplier1"}
        //{"bidtime": "2020-04-15 08:08:00.000", "item" : "B", "price": 2, "supplier_id": "supplier3"}
        tEnv.executeSql(
                "CREATE TABLE Bid (\n" +
                        "  `item` STRING,  \n" +
                        "  `price` DOUBLE,  \n" +
                        "  `supplier_id` STRING, \n" +
                        "  `bidtime` TIMESTAMP(3),  \n" +
                        "  WATERMARK FOR bidtime AS bidtime - INTERVAL '0' SECOND" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'demo1-json',\n" +
                        "  'properties.bootstrap.servers' = 'node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092',\n" +
                        "  'properties.group.id' = 'testGroup',\n" +
                        "  'scan.startup.mode' = 'earliest-offset',\n" +
                        "  'format' = 'json', \n" +
                        "  'json.ignore-parse-errors' = 'true'" +
                        ")"
        );

        //TableResult tableResult0 = tEnv.executeSql("desc kafka_event");
        //tableResult0.print();

        //TableResult tableResult = tEnv.executeSql("select * from kafka_event");
        //tableResult.print();

        tEnv.executeSql(
                "CREATE TEMPORARY VIEW v_temp AS SELECT supplier_id, SUM(price) as price, COUNT(*) as cnt, TUMBLE_START(bidtime, INTERVAL '10' MINUTES) as window_start, " +
                        "TUMBLE_END(bidtime, INTERVAL '10' MINUTES) as window_end " +
                        "FROM Bid GROUP BY TUMBLE(bidtime, INTERVAL '10' MINUTES), supplier_id"
        );

        TableResult tableResult = tEnv.executeSql(
                "SELECT * FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY price DESC) as rownum FROM v_temp) WHERE rownum <= 3"
        );


        tableResult.print();




    }
}
